package p.ithorns.framework.common.utils;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.*;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;

/**
 * CompletableUtil
 *
 * @author HtL
 * @date 2024/8/1 16:54
 * @since 1.0.0
 */
public class CompletableUtil {

    private final static Logger log = LoggerFactory.getLogger(CompletableUtil.class);

//    private void multiFuture() {
//        return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
//                .thenApply(v -> {
//                    return futures.stream()
//                            .map(CompletableFuture::join)
//                            .flatMap(List::stream)
//                            .collect(Collectors.toList());
//                })
//                .join();
//    }

    public static void main(String[] args) throws Exception {
        // 异常被吞噬
        //  unThrow();

         // exceptionally
        exceptionally();

        // 异常处理丢失
        // missError();

        //throwWithComplete();

        System.exit(0);
    }

    private static void exceptionally() {
        CompletableFuture.runAsync(() -> {
            int i = 1 / 0;
        }).exceptionally(t -> {
            t.printStackTrace();
            return null;
        });
    }

    private static void unThrow() {
        try {
            try {
                CompletableFuture.runAsync(() -> {
                    int i = 1 / 0;
                }); //  不加join,无法捕获异常
                 // .join();
            } catch (Exception e) {
                log.error("异常信息: " + e.getMessage(), e);
                throw new RuntimeException(e.getMessage());
            }
            System.out.println("操作成功");
        } catch (Exception e) {
            System.out.println("操作失败:" + e.getMessage());
        }
    }

    public void handleErrorWithMessenger() {
        CompletableFuture<String> messenger = new CompletableFuture<>();
        Queue<CompletableFuture<String>> futures = new ConcurrentLinkedQueue<>();

        // 异常处理
        messenger.exceptionally(e -> {
            log.error("异步存储文件执行失败.", e);
            // 取消所有未完成的任务
            for (CompletableFuture<String> future : futures) {
                if (future.isDone()) {
                    future.cancel(true);
                }
            }
            return null;
        });

        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();

    }

    public void handleErrorWhenComplete() {
        Queue<CompletableFuture<String>> futures = new ConcurrentLinkedQueue<>();
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
                .whenComplete((aVoid, ex) -> log.info("任务已完成"));

        final List<String> result = new ArrayList<>(futures.size());
        futures.stream()
                .map(CompletableFuture::join)
                .filter(Objects::nonNull)
                .forEach(result::add);

        System.out.println(result);
    }

    /**
     * 异常被吞噬
     */
    private static void swallowError() throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            throw new RuntimeException("Oops!");
        });

        CompletableFuture<String> result = future.thenApply(i -> "Success: " + i);

        result.join(); // 此处不会抛出异常

        System.out.println(result.get());
    }

    private static void missError() {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            throw new RuntimeException("Oops!");
        });

        CompletableFuture<String> result = future.exceptionally(ex -> {
            System.out.println("Error occurred: " + ex);
            return "Default Value";
        });

        result.join(); // 此处不会输出错误信息
    }

    private static void throwWithComplete() {
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            throw new RuntimeException("Oops!");
        });

        CompletableFuture<String> result = future.thenApply(i -> "Success: " + i)
                .whenComplete((res, ex) -> {
                    if (ex != null) {
                        System.out.println("Error occurred: " + ex);
                    }
                });

        result.join(); // 此处会输出错误信息
    }


    /**
     * 创建并行任务并执行
     */
    public <S, T> List<T> join(Collection<S> list, Function<S, T> api, Executor executor, BiFunction<Throwable, S, T> exHandler) {
        //规整所有任务
        List<CompletableFuture<T>> futures = list.stream()
                .map(item -> this.createFuture(() ->
                        api.apply(item), executor, e -> exHandler.apply(this.unwrapCompletionException(e), item)))
                .toList();
        return CompletableFuture.allOf(futures.toArray(new CompletableFuture<?>[]{}))
                .thenApply(f -> futures.stream()
                        .map(CompletableFuture::join)
                        .filter(Objects::nonNull)
                        .collect(Collectors.toList()))
                .join();
    }

    /**
     * 创建CompletableFuture任务
     */
    public <T> CompletableFuture<T> createFuture(Supplier<T> supplier, Executor executor,
                                                 Function<Throwable, T> exHandler) {
        if (null == executor) {
            return CompletableFuture.supplyAsync(supplier).exceptionally(exHandler);
        }
        return CompletableFuture.supplyAsync(supplier, executor).exceptionally(exHandler);
    }


    /**
     * 提取真正的异常
     * <p>
     * CompletableFuture抛出的往往不是真正的异常
     *
     * @param throwable 异常
     * @return 真正的异常
     */
    public Throwable unwrapCompletionException(Throwable throwable) {
        if (throwable instanceof CompletionException || throwable instanceof ExecutionException) {
            if (throwable.getCause() != null) {
                return throwable.getCause();
            }
        }
        return throwable;
    }

}